package com.swak.reactivex.transport.udp;

import java.io.IOException;

import org.springframework.util.Assert;

import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.Codec;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.NettyChannel;
import com.swak.reactivex.transport.exception.CodecException;
import com.swak.registry.URL;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.DatagramPacket;
import io.netty.handler.codec.DecoderException;
import lombok.Getter;

/**
 * Netty Codec.
 * 
 * @author lifeng
 */
public final class NettyCodec implements Codec {

	@Getter
	private final ChannelHandler encoder = new InternalEncoder();
	@Getter
	private final ChannelHandler decoder = new InternalDecoder();
	@Getter
	private final URL url;
	@Getter
	private final com.swak.reactivex.transport.ChannelHandler handler;

	private final Codec codec;

	public NettyCodec(Codec codec, URL url, com.swak.reactivex.transport.ChannelHandler handler) {
		Assert.isTrue(codec != null, "codec == null");
		Assert.isTrue(url != null, "url == null");
		Assert.isTrue(handler != null, "handler == null");
		this.codec = codec;
		this.url = url;
		this.handler = handler;
	}

	/**
	 * 编码
	 * 
	 * @param channel
	 * @param buffer
	 * @param message
	 * @throws IOException
	 */
	public void encode(Channel channel, ByteBuf buffer, Object message) throws CodecException {
		this.codec.encode(channel, buffer, message);
	}

	/**
	 * 解码
	 * 
	 * @param channel
	 * @param buffer
	 * @return
	 * @throws IOException
	 */
	public Object decode(Channel channel, ByteBuf buffer) throws CodecException {
		return this.codec.decode(channel, buffer);
	}

	/**
	 * 编码
	 * 
	 * @author lifeng
	 *
	 */
	class InternalEncoder extends MessageToDatagramEncoder {

		@Override
		protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
			NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
			NettyCodec.this.encode(channel, out, msg);
		}
	}

	/**
	 * 解码
	 * 
	 * @author lifeng
	 */
	class InternalDecoder extends ChannelInboundHandlerAdapter {

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
			Assert.isTrue(message instanceof DatagramPacket, "message must instanceof DatagramPacket.");
			NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
			DatagramPacket packet = (DatagramPacket) message;
			ByteBuf buffer = packet.content();
			Object msg = null;
			try {
				msg = NettyCodec.this.decode(channel, buffer);

				// 这里由于DNS的影响，导致处理速度慢
				// 其实也是无关紧要的代码
				if (msg instanceof Message) {
					Message realMessage = (Message) msg;
					realMessage.setAddress(packet.sender());
				}
			} catch (DecoderException e) {
				throw e;
			} catch (Exception e) {
				throw new DecoderException(e);
			} finally {

				// 释放资源
				buffer.release();

				// 后续处理
				if (msg != null) {
					ctx.fireChannelRead(msg);
				}
			}
		}
	}
}
